package com.iot.service.mqtt.service.session;

import com.iot.service.mqtt.model.TlMqttSession;
import com.iot.service.mqtt.service.StoreStrategy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * @Author: hszhou
 * @Date: 2025/1/8 9:11
 * @Description: 默认的session处理实现类 使用本地
 */
@Service
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.REDIS)
@RequiredArgsConstructor
@Slf4j
public class RedisSessionServiceImpl implements ISessionService {


    private final ReactiveRedisTemplate<String, Object> reactiveRedisTemplate;

    private static final String PREFIX = "session:";

    @Override
    public Mono<Boolean> save(TlMqttSession session) {
        String clientId = session.getClientId();
        return reactiveRedisTemplate.opsForValue().set(PREFIX + clientId, session)
                .doOnError(e -> log.error("save byClientId failed for key: {}", PREFIX+clientId, e));
    }

    @Override
    public Mono<TlMqttSession> find(String clientId) {
        return reactiveRedisTemplate.opsForValue().get(PREFIX + clientId)
                .doOnError(e -> log.error("find byClientId failed for key: {}", PREFIX+clientId, e))
                .cast(TlMqttSession.class);
    }

    @Override
    public Mono<Boolean> clear(String clientId) {
        return reactiveRedisTemplate.opsForValue().delete(PREFIX + clientId)
                .doOnError(e -> log.error("clear byClientId failed for key: {}", PREFIX+clientId, e));
    }


    @Override
    public Flux<String> clients() {

        return reactiveRedisTemplate.keys(PREFIX + "*")
                .doOnError(e -> log.error("clear byClientId failed for key: {}", PREFIX + "*", e));
    }
}
